package net.imagine.backend.mockrmi.remote.provider.netty.client;

import net.imagine.backend.mockrmi.remote.client.RequestSender;
import net.imagine.backend.mockrmi.remote.protocol.RequestBean;
import net.imagine.backend.mockrmi.remote.protocol.ResponseBean;
import net.imagine.backend.mockrmi.util.Constants;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;

/**
 * Netty实现的RequestSender
 * 
 * @author wuqh
 * @see RequestSender
 * @see NettyClientContext
 */
public class NettyRequestSender implements RequestSender {
	private static final Log logger = LogFactory.getLog(NettyRequestSender.class);
	private final boolean retry;
	private final boolean doClose;
	private final NettyClientContext context = new NettyClientContext();

	public NettyRequestSender() {
		this(false, false);
	}

	/**
	 * 构造函数
	 * 
	 * @param retry
	 *            失败后是否重试
	 * @param doClose
	 *            完成后是否关闭
	 */
	public NettyRequestSender(boolean retry, boolean doClose) {
		this.retry = retry;
		this.doClose = doClose;
	}

	@Override
	public ResponseBean sendRequest(RequestBean req, String domain, int port) throws Exception {
		Object lock = new Object();
		Channel channel = context.createChannel(domain, port);
		if (retry) {// 未连接则重试
			int connTryTimes = 1;
			synchronized (lock) {
				while (connTryTimes < Constants.CLIENT_RETRY_TIMES && (channel == null || !channel.isConnected())) {
					logger.warn("connect to server failed, it will retry after " + Constants.CLIENT_RETRY_DEALY + "ms");
					lock.wait(Constants.CLIENT_RETRY_DEALY);
					channel = context.createChannel(domain, port);
					connTryTimes++;
				}
			}
			if (connTryTimes >= Constants.CLIENT_RETRY_TIMES) {
				logger.warn("connect to server failed afer retry " + Constants.CLIENT_RETRY_TIMES + "times");
				throw new Exception("Connect to server failed afer retry " + Constants.CLIENT_RETRY_TIMES + " times");
			}
		}
		ChannelFuture writeFuture = channel.write(req);
		writeFuture.awaitUninterruptibly(Constants.CLIENT_WRITE_TIMEOUT);
		if (retry) {// 未写成功重试
			int writeTryTimes = 1;
			synchronized (lock) {
				while (writeTryTimes < Constants.CLIENT_RETRY_TIMES && !writeFuture.isSuccess()) {
					logger.warn("write to server failed, it will retry after " + Constants.CLIENT_RETRY_DEALY + "ms");
					lock.wait(Constants.CLIENT_RETRY_DEALY);
					writeFuture = channel.write(req);
					writeFuture.awaitUninterruptibly(Constants.CLIENT_WRITE_TIMEOUT);
					writeTryTimes++;
				}
			}
			if (writeTryTimes >= Constants.CLIENT_RETRY_TIMES) {
				logger.warn("write to server failed afer retry " + Constants.CLIENT_RETRY_TIMES + "times");
				throw new Exception("Write to server failed afer retry " + Constants.CLIENT_RETRY_TIMES + " times");
			}
		}
		
		Object message = null;
		long readWaited = 0L;
		
		while (message == null && readWaited < Constants.CLIENT_READ_TIMEOUT) {
			Thread.sleep(1L);
			message = channel.getPipeline().getContext(channel.getPipeline().getLast()).getAttachment();
			readWaited++;
		}
		if(readWaited >= Constants.CLIENT_READ_TIMEOUT) {
			throw new Exception("Read from server failed afert " + Constants.CLIENT_READ_TIMEOUT + " ms");
		}
		if (doClose) {
			context.releaseChannel(channel);
		}

		ResponseBean response = null;
		if (message instanceof ResponseBean) {
			response = (ResponseBean) message;
		} else {
			logger.warn("Bad Response! response is not instance of ResponseBean");
			throw new Exception("Bad Response! response is not instance of ResponseBean");
		}
		if (response.getReturnValue() == null) {
			logger.warn("method returns null");
		}
		if (response.getReturnValue() instanceof Throwable) {
			logger.warn("method returns an exception");
			Throwable e = (Throwable) response.getReturnValue();
			throw new Exception(e);
		}
		return response;
	}
}
